package jpf.nextsearch.processing.invoker.impl;

import java.util.Collection;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Stack;
//XXXX koment + logi !!!!!!
import jpf.nextsearch.processing.invoker.ProcessorInvoker;

/**
 * @author <a href="mailto:jacek.pruszkowski@gmail.com">Jacek Pruszkowski</a>
 */
public class ConcurentProcessorInvoker<T> extends ProcessorInvoker<T>{
	private int maxThreadsCount = 5;
	
	private int currentRunnersCount = 0;
	private int runningRunners = 0;
	private boolean active = true;
	
	private Stack<ProcessingRunner<T>> runners = new Stack<ProcessingRunner<T>>();
	
	@Override
	public void invokeProcessor(Collection<T> data) throws Exception {
		Queue<T> dataQueue = new LinkedList<T>(data);
		
		if (isLoggingInfo()) {
			getLogger().info("Invoking concurent processing for processor : "+getProcessor().getClass().getName());
		}
		
		synchronized (this) {
			T item = null;
			while ((item = dataQueue.poll()) != null && isActive()) {
				ProcessingRunner<T> runner = null;
				
				if (isLoggingDebug()) {
					getLogger().debug("Processing item: "+item.toString());
				}

				while ((runner = getRunner()) == null) {
					if (isLoggingDebug()) {
						getLogger().debug("Waiting for idle runner ...");
					}
					wait();
				}
				runner.setData(item);
				notifyAll();

				if (!runner.isRunning()) {
					new Thread(runner).start();
				}
				
				++runningRunners;
			}
			
			while (runningRunners > 0) {
				if (isLoggingDebug()) {
					getLogger().debug("Waiting for all runners to stop processing ...");
				}
				wait();
			}
		}
		
		if (isLoggingInfo()) {
			getLogger().info("Concurent processing for processor : "+getProcessor().getClass().getName() +" finished");
		}
		
	}
	
	private synchronized ProcessingRunner<T> getRunner() {
		if (runners.size() > 0) {
			return runners.pop();
		}
		
		if (currentRunnersCount < getMaxThreadsCount()) {
			++currentRunnersCount;
			return new ProcessingRunner<T>(this);
		}
		
		return null;
	}
	
	private synchronized void returnRunner(ProcessingRunner<T> runner) {
		runner.setData(null);
		runners.push(runner);
		--runningRunners;
		notifyAll();
	}
	
	private synchronized void removeRunner(ProcessingRunner<T> runner) {
		--runningRunners;
		--currentRunnersCount;
		notifyAll();
	}	
	
	private static final class ProcessingRunner<T> implements Runnable {
		private T data;
		private ConcurentProcessorInvoker<T> invoker;
		private boolean running;
		
		public ProcessingRunner(ConcurentProcessorInvoker<T> invoker) {
			this.invoker = invoker;
		}

		@Override
		public void run() {
			running = true;
			while (true) {
				try {
					
					if (data != null) {
						invoker.getProcessor().process(data);
					}
					
					invoker.returnRunner(this);
	
					synchronized (invoker) {
						while (invoker.isActive() && data == null) {
							try {
								invoker.wait();
							} catch (InterruptedException e) {
								e.printStackTrace();
								invoker.removeRunner(this);
								return; 
							}
						}
						
						if (!invoker.isActive()) {
							invoker.removeRunner(this);
							return;
						}
					}
	
					} catch (Exception e) {
						e.printStackTrace();
						invoker.removeRunner(this);
						return;
					}
								
			}
		}

		public void setData(T data) {
			this.data = data;
		}

		public boolean isRunning() {
			return running;
		}
		
	}
	
	public void setMaxThreadsCount(int maxThreadsCount) {
		this.maxThreadsCount = maxThreadsCount;
	}

	public int getMaxThreadsCount() {
		return maxThreadsCount;
	}

	public synchronized void setActive(boolean active) {
		this.active = active;
		notifyAll();
	}

	public boolean isActive() {
		return active;
	}
		

}
